package com.platform.system.mqtt;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import java.util.concurrent.*;

/**
 * @author Huangkai
 */
public class AdvancedMqttDeviceClient implements MqttCallback {

    private static final String BROKER = "tcp://localhost:1883";
    private static final String CLIENT_ID = "PHNsn4rWQ4O2d1Ue6p5kcnlirP3vtVvBtnXypUnLRbAYw98CS94gUPJFutFRsMgo";
    //发布命令主题
    private static final String PUB_TOPIC = "device/data";
    //订阅命令主题
    private static final String SUB_TOPIC = "device/info";

    private MqttClient client;
    private ScheduledExecutorService scheduler;
    private BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();

    public void start() {
        scheduler = Executors.newSingleThreadScheduledExecutor();
        connect();
        // 启动消息发送线程
        scheduler.scheduleAtFixedRate(this::sendQueuedMessages, 0, 1, TimeUnit.SECONDS);
    }

    private void connect() {
        try {
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setAutomaticReconnect(true);
            options.setConnectionTimeout(10);
            options.setKeepAliveInterval(10);

            client = new MqttClient(BROKER, CLIENT_ID, new MemoryPersistence());
            client.setCallback(this);
            client.connect(options);

            // 订阅命令主题
            client.subscribe(SUB_TOPIC, 1);
            System.out.println("MQTT连接成功，已订阅主题: " + SUB_TOPIC);
        } catch (MqttException e) {
            System.err.println("连接失败，5秒后重试...");
            scheduler.schedule(this::connect, 5, TimeUnit.SECONDS);
        }
    }

    public void sendMessage(String message) {
        messageQueue.offer(message);
    }

    private void sendQueuedMessages() {
        while (!messageQueue.isEmpty()) {
            String message = messageQueue.poll();
            if (message != null && client != null && client.isConnected()) {
                try {
                    MqttMessage mqttMessage = new MqttMessage(message.getBytes());
                    mqttMessage.setQos(1);
                    client.publish(PUB_TOPIC, mqttMessage);
                    System.out.println("消息已发送: " + message);
                } catch (MqttException e) {
                    System.err.println("消息发送失败，重新加入队列");
                    messageQueue.offer(message);
                    break;
                }
            }
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        System.err.println("连接断开: " + cause.getMessage());
        scheduler.schedule(this::connect, 5, TimeUnit.SECONDS);
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
        System.out.println("收到命令: " + new String(message.getPayload()));
        // 处理服务器下发的命令
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // 消息发送完成回调
    }

    public void shutdown() {
        try {
            if (client != null && client.isConnected()) {
                client.disconnect();
                client.close();
            }
            scheduler.shutdown();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        AdvancedMqttDeviceClient device = new AdvancedMqttDeviceClient();
        device.start();

        // 模拟发送传感器数据
        for (int i = 0; i < 10; i++) {
            String data = String.format("{\"temp\":%.1f,\"humidity\":%d}",
                    Math.random() * 10 + 20,
                    (int)(Math.random() * 30 + 50));
            device.sendMessage(data);
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        device.shutdown();
    }
}